package com.hhf.rrd.udf;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * udf 测试
 *
 * @author huanghaifeng15
 * @date 2022/2/11 14:49
 **/
public class UdfApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> dataStreamSource = env.socketTextStream("localhost", 9999);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // register function
        tableEnv.createTemporarySystemFunction("hashCode", HashCodeFunction.class);

        tableEnv.createTemporaryView("access",dataStreamSource,$("num"));

        Table resultTable = tableEnv.sqlQuery("select num, hashCode(num) from access");
        tableEnv.toAppendStream(resultTable, Row.class).print();

        env.execute();
    }
}
